package sse.util;

import org.apache.hc.client5.http.classic.methods.HttpPost;
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
import org.apache.hc.client5.http.impl.classic.HttpClients;
import org.apache.hc.core5.http.ContentType;
import org.apache.hc.core5.http.io.entity.StringEntity;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.scheduler.Schedulers;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.atomic.AtomicBoolean;

public class Caller {

    public static Flux<String> callByStream(String url, String postData) {
        return Flux.create(sink -> {
            // https://blog.csdn.net/gitblog_00278/article/details/148507634
            // single    新开一个线程执行流式任务
            // immediate 在当前线程执行流式任务
            Schedulers.immediate().schedule(() -> {
                CloseableHttpClient client = HttpClients.createDefault();
                HttpPost httpPost = new HttpPost(url);

                try {
                    // 设置POST参数
                    StringEntity entity = new StringEntity(postData, ContentType.APPLICATION_FORM_URLENCODED);
                    httpPost.setEntity(entity);

                    // 设置SSE所需的请求头
                    httpPost.setHeader("Accept", "text/event-stream");
                    httpPost.setHeader("Cache-Control", "no-cache");

                    // 执行请求
                    try (CloseableHttpResponse response = client.execute(httpPost)) {
                        // 检查响应状态
                        if (response.getCode() != 200) {
                            sink.error(new RuntimeException("HTTP error: " + response.getCode()));
                            return;
                        }

                        // 获取响应流
                        InputStream inputStream = response.getEntity().getContent();
                        BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8));

                        // 读取SSE流
                        String line;
                        AtomicBoolean isConnected = new AtomicBoolean(true);

                        sink.onCancel(() -> {
                            isConnected.set(false);
                            try {
                                client.close();
                            } catch (IOException e) {
                                // 忽略关闭异常
                            }
                        });

                        while (isConnected.get() && (line = reader.readLine()) != null) {
                            if (line.startsWith("data:")) {
                                // 提取数据部分
                                String data = line.substring("data:".length()).trim();
                                if (!data.isEmpty()) {
                                    sink.next(data);
                                }
                            } else if (line.isEmpty()) {
                                // 空行表示一个事件结束
                                // 可以在这里处理完整的事件
                            }
                            // 忽略注释行(以:开头)和其他行
                        }

                        // 流结束
                        sink.complete();
                    }
                } catch (Exception e) {
                    sink.error(e);
                } finally {
                    try {
                        client.close();
                    } catch (IOException e) {
                        // 忽略关闭异常
                    }
                }
            });
        }, FluxSink.OverflowStrategy.BUFFER);
    }
}
